package function

import com.cw.realtime.common.bean.TrafficHomeDetailPageViewBean
import com.cw.realtime.common.util.DateFormatUtil._
import io.circe.JsonObject
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

class KPFunc4HomeDetail extends KeyedProcessFunction[String, JsonObject, TrafficHomeDetailPageViewBean] {
  private type FuncType = KeyedProcessFunction[String, JsonObject, TrafficHomeDetailPageViewBean]

  private val homeStateDesc = new ValueStateDescriptor("home_last_login", classOf[String])
  private val detailStateDesc = new ValueStateDescriptor("good_detail_last_login", classOf[String])

  homeStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build())
  detailStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build())

  private var homeState = Option.empty[ValueState[String]]
  private var detailState = Option.empty[ValueState[String]]

  override def open(parameters: Configuration): Unit = {
    homeState = Some(getRuntimeContext.getState(homeStateDesc))
    detailState = Some(getRuntimeContext.getState(detailStateDesc))
  }

  override def processElement(value: JsonObject, ctx: FuncType#Context, out: Collector[TrafficHomeDetailPageViewBean]): Unit = {
    for {
      ts <- value("ts")
      ts <- ts.asNumber.get.toLong
      page <- value("page")
      page <- page.asObject
      pageId <- page("page_id")
      pageId <- pageId.asString
      homeState <- homeState
      detailState <- detailState

      loginDt = ts.toDate
      homeLastLogin = homeState.value
      detailLastLogin = detailState.value
    } {
      val homeUv =
        if ("home" == pageId && (homeLastLogin == null || homeLastLogin != loginDt)) 1
        else 0

      val detailUv =
        if ("good_detail" == pageId && (detailLastLogin == null || detailLastLogin != loginDt)) 1
        else 0

      homeState.update(loginDt)

      if (homeUv != 0 || detailUv != 0)
        out.collect(TrafficHomeDetailPageViewBean(homeUv, detailUv, ts))
    }
  }

}